package zmqtt

import (
	"context"
	"crypto/tls"
	"crypto/x509"
	"errors"
	"fmt"
	MQTT "github.com/eclipse/paho.mqtt.golang"
	"os"
	"time"
)

type OnConnected func(cli MQTTClient)
type OnconnectLost func(client MQTTClient, err error)
type OnMessage func(cli MQTTClient, msg Message)
type Message MQTT.Message

type MQTTClient struct {
	opts               *MQTT.ClientOptions //
	cli                MQTT.Client         //
	user               string              //
	pwd                string              //
	addr               string              //
	cliId              string              //
	caFile             string              //
	certFile           string              //
	keyFile            string              //
	isConnected        bool                //
	reConnected        bool                // 断线自动重连
	connectHandler     OnConnected         // 连接事件
	connectLostHandler OnconnectLost       // 断掉事件
	messageHandler     OnMessage           // 消息事件
}

func NewMQTTClient(clientId, addr string, port int, reconnect bool) *MQTTClient {
	var c MQTTClient
	c.cliId = clientId
	c.addr = addr
	if c.opts == nil {
		c.opts = MQTT.NewClientOptions()
	}

	c.opts.AddBroker(fmt.Sprintf("tcp://%s:%d", addr, port))
	c.opts.SetClientID(clientId)
	c.reConnected = reconnect
	//c.cli = MQTT.NewClient(c.opts)
	return &c
}

// ConnectPwd 密码连接
func (c *MQTTClient) ConnectPwd(user, pwd string) (e error) {
	c.pwd = pwd
	c.user = user
	c.opts.SetUsername(user)
	c.opts.SetPassword(pwd)
	c.opts.SetAutoReconnect(c.reConnected)

	return c.connect()
}

// ConnectTls tsl连接
func (c *MQTTClient) ConnectTls(caFile, certFile, keyFile string) (e error) {
	c.caFile = caFile
	c.certFile = certFile
	c.keyFile = keyFile
	if e = c.setTls(caFile, certFile, keyFile); e != nil {
		return
	}
	return c.connect()
}

// Subscribe 订阅
func (c *MQTTClient) Subscribe(topic string) error {
	if token := c.cli.Subscribe(topic, 0, c.onMessageEvent); token.Wait() && token.Error() != nil {
		return token.Error()
	}
	return nil
}

// Unsubscribe 取消订单
func (c *MQTTClient) Unsubscribe(topic string) error {

	if token := c.cli.Unsubscribe(topic); token.Wait() && token.Error() != nil {
		return token.Error()
	}
	return nil
}

// Publish 发布
func (c *MQTTClient) Publish(topic, content string) error {
	if !c.isConnected {
		return errors.New("mqtt client not connected")
	}
	if token := c.cli.Publish(topic, 0, false, content); token.Wait() && token.Error() != nil {
		return token.Error()
	}
	return nil
}

// DisConnect 断开连接
func (c *MQTTClient) DisConnect() {
	c.cli.Disconnect(0)
}

// SetConnectLostEvent 设置离线事件
func (c *MQTTClient) SetConnectLostEvent(ev OnconnectLost) {
	c.connectLostHandler = ev
}

// SetConnectedEvent 设置连接事件
func (c *MQTTClient) SetConnectedEvent(ev OnConnected) {
	c.connectHandler = ev
}

// SetMessageEvent 设置消息事件
func (c *MQTTClient) SetMessageEvent(ev OnMessage) {
	c.messageHandler = ev
}

// SetWill 设置遗嘱
func (c *MQTTClient) SetWill(topic string) {
	payload := "{\"clientId\":\"" + c.cliId + "\",\"time\":\"" + time.Now().Format(time.DateTime) + "\"}"
	c.opts.SetWill(topic, payload, 1, true)
}

// =================================================私有=================================================================

// 连接
func (c *MQTTClient) connect() error {
	//if c.cli.IsConnected() {
	//	return errors.New("already connected")
	//}
	c.opts.OnConnect = c.onConnectedEvent
	c.opts.OnConnectionLost = c.onConnectLostEvent

	//c.opts.SetDialer(&net.Dialer{Timeout: 5 * time.Second})
	//c.opts.SetConnectTimeout(5 * time.Second)
	c.opts.SetAutoReconnect(c.reConnected)
	c.opts.SetConnectRetry(c.reConnected)

	c.cli = MQTT.NewClient(c.opts)
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	//if token := c.cli.Connect(); token.Wait() && token.Error() != nil {
	//	return token.Error()
	//}
	token := c.cli.Connect()
	select {
	case <-token.Done():
		if token.Error() != nil {
			c.cli.Disconnect(0)
			return token.Error()
		}
	case <-ctx.Done():
		c.cli.Disconnect(0)
		return errors.New("connect timeout")
	}
	return nil
}

// SetTls 设置TLS信息
func (c *MQTTClient) setTls(caFile, certFile, keyFile string) error {
	caPool := x509.NewCertPool()
	ca, e := os.ReadFile(caFile)
	if e != nil {
		return errors.New("error reading ca file")
	}
	if !caPool.AppendCertsFromPEM(ca) {
		return errors.New("error parsing ca file")
	}
	cert, e := tls.LoadX509KeyPair(certFile, keyFile)
	if e != nil {
		return errors.New("error loading certificate")
	}

	tlsConfig := &tls.Config{
		RootCAs:            caPool,
		Certificates:       []tls.Certificate{cert},
		ClientAuth:         tls.NoClientCert,
		ClientCAs:          nil,
		MinVersion:         tls.VersionTLS10,
		MaxVersion:         tls.VersionTLS12,
		InsecureSkipVerify: false,
	}
	c.opts.SetTLSConfig(tlsConfig)
	return c.connect()
}

// 连接事件
func (c *MQTTClient) onConnectedEvent(cli MQTT.Client) {
	c.isConnected = true
	if c.connectHandler != nil {
		c.connectHandler(MQTTClient{cli: cli})
	}
}

// 离线事件
func (c *MQTTClient) onConnectLostEvent(cli MQTT.Client, err error) {
	c.isConnected = false
	if c.connectLostHandler != nil {
		c.connectLostHandler(MQTTClient{cli: cli}, err)
	}
}

// 消息事件
func (c *MQTTClient) onMessageEvent(client MQTT.Client, msg MQTT.Message) {
	if c.messageHandler != nil {
		c.messageHandler(MQTTClient{cli: client}, msg)
	}
}
